package com.bxfy.rabbit.producer.broker;

import com.bfxy.rabbit.api.Message;
import com.bfxy.rabbit.api.MessageType;
import com.bxfy.rabbit.producer.constant.BrokerMessageConst;
import com.bxfy.rabbit.producer.constant.BrokerMessageStatus;
import com.bxfy.rabbit.producer.entity.BrokerMessage;
import com.bxfy.rabbit.producer.service.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author caryguan
 * @version V1.0
 * @Package com.bxfy.rabbit.producer.broker
 * @date 2020/9/30 19:32
 */

@Slf4j
@Component //spring接管
public class RabbitBrokerImpl implements RabbitBroker {

    @Autowired
    private MessageStoreService messageStoreService;

    @Autowired
    private RabbitTemplateContainer rabbitTemplateContainer;

    @Override
    public void rapidSend(Message message) {
        message.setMessageType(MessageType.RAPID);
        sendKernel(message);
    }

    /**
     * 发送消息核心方法不对外暴露,使用异步的线程池进行发送消息
     *
     * @param message
     */
    private void sendKernel(Message message) {
        AsyncBaseQueue.submit((Runnable) () -> {
            String topic = message.getTopic();
            String routingKey = message.getTopic();
            CorrelationData correlationData = new CorrelationData(String.format("%s#%s",
                    message.getMessageId(),
                    System.currentTimeMillis()));
            RabbitTemplate redisTemplate = rabbitTemplateContainer.getTemplate(message);
            redisTemplate.convertAndSend(topic, routingKey, message, correlationData);
            log.info("========================================================");
            log.info("#RabbitBrokerImpl.sendKernel# send to rabbitMQ, messageId: {}", message.getMessageId());
        });

    }

    @Override
    public void confirmSend(Message message) {
        message.setMessageType(MessageType.CONFIRM);
        sendKernel(message);
    }

    @Override
    public void reliantSend(Message message) {

        //为了确保万无一失,再发之前做一个消息,类型的验证
        message.setMessageType(MessageType.RELIANT);

        //1.首先new一个BrokerMessage并设置信息发送状态为0,nextRetry,tryCount不变因为默认为0就是我们要的,nextRetry
        Date now = new Date();

        BrokerMessage brokerMessage = new BrokerMessage();
        brokerMessage.setStatus(BrokerMessageStatus.SENDING.getCode());
        //我们设置一分钟重试一次的机制
        brokerMessage.setNextRetry(DateUtils.addMinutes(now, BrokerMessageConst.TIMEOUT));
        brokerMessage.setCreateTime(now);
        brokerMessage.setUpdateTime(now);
        brokerMessage.setMessage(message);

        messageStoreService.insert(brokerMessage);


        //2.执行真正的发送消息逻辑
        sendKernel(message);


    }

    @Override
    public void sendMessages(Message message) {

    }
}